package bullgo

import (
	"math"
	"time"
)

// Process ... jobHander JobHandler, concurrency int
func (queue *Queue) Process(jobHander JobHandler, concurrency int) {
	// 处理定时任务
	queue.setGuardianTicker()
	// 处理卡住的任务
	queue.setMoveUnlockedJobsToWaitTicker()
	queue.moveUnlockedJobsToWait()
	// 处理定时任务
	now := time.Now().UnixNano() / 1e6
	queue.delayUpdate(now)
	jobChan := queue.getNextJobs()
	for i := 0; i < concurrency; i++ {
		go func() {
			done := make(chan int)
			defer close(done)
			for job := range jobChan {
				queue.TickerManger.Set(
					"extendLock:"+job.ID,
					time.Duration(queue.queueSettings.LockRenewTime)*time.Millisecond,
					func() {
						extendLock(queue, job.ID)
					},
				)
				go jobHander(job, func(err error, returnValue interface{}) {
					if err != nil {
						job.MoveToFailed(err, false)
					} else {
						job.MoveToCompleted(returnValue, false)
					}
					queue.TickerManger.Clear("extendLock:" + job.ID)
					done <- 1
				})
				<-done
			}
		}()
	}
}

func (queue *Queue) updateDelayTimer(newDelayedTimestamp int64) {
	if newDelayedTimestamp < queue.delayedTimestamp {
		now := time.Now().UnixNano() / 1e6
		queue.delayedTimestamp = newDelayedTimestamp
		nextDelayedJob := newDelayedTimestamp - now
		delay := int64(0)
		if nextDelayedJob > 0 {
			delay = nextDelayedJob
		}
		if delay > 0 {
			queue.TimerManger.Set("updateDelayTimer", time.Duration(delay)*time.Millisecond, func() {
				queue.delayUpdate(now)
			})
		} else {
			queue.delayUpdate(now)
		}
	}
}

func (queue *Queue) delayUpdate(now int64) {
	var nextTimestamp int64 = math.MaxInt64
	queue.delayedTimestamp = math.MaxInt64
	result, _ := updateDelaySet(queue, now).Result()
	if result != nil {
		nextTimestamp = result.(int64)
		if nextTimestamp < now {
			nextTimestamp = now
		}
	}
	queue.updateDelayTimer(nextTimestamp)
}

func (queue *Queue) setGuardianTicker() {
	queue.TickerManger.Set("setGuardianTicker", time.Duration(queue.queueSettings.GuardInterval)*time.Millisecond, func() {
		now := time.Now().UnixNano() / 1e6
		if queue.delayedTimestamp < now ||
			queue.delayedTimestamp-now > queue.queueSettings.GuardInterval {
			queue.delayUpdate(now)
		}
	})
}

func (queue *Queue) moveUnlockedJobsToWait() {
	moveUnlockedJobsToWait(queue).Result()
}

func (queue *Queue) setMoveUnlockedJobsToWaitTicker() {
	queue.TickerManger.Set("setMoveUnlockedJobsToWaitTicker", time.Duration(queue.queueSettings.StalledInterval)*time.Millisecond, func() {
		queue.moveUnlockedJobsToWait()
	})
}

func (queue *Queue) getNextJobs() chan *Job {
	jobChan := make(chan *Job, 1)
	go getNextJob(queue, jobChan)
	return jobChan
}

func getNextJob(queue *Queue, jobChan chan *Job) {
	var jobId string
	var err error
	var val interface{}
	if queue.Drained {
		jobId, err = queue.ExtendRedis.Client.BRPopLPush(
			queue.ExtendRedis.Ctx,
			queue.queueKeys.Wait,
			queue.queueKeys.Active,
			time.Duration(queue.queueSettings.DrainDelay)*time.Second,
		).Result()
	}
	if err == nil {
		val, _ = moveToActive(queue, jobId).Result()
	} else {
		val, _ = moveToActive(queue, "").Result()
	}

	if val != nil {
		jobMap := map[string]string{}
		data := val.([]interface{})
		jobID := data[1].(string)
		jobMap["jobID"] = jobID
		jobData := data[0].([]interface{})
		for i := 0; i < len(jobData); i = i + 2 {
			jobMap[jobData[i].(string)] = jobData[i+1].(string)
		}
		job, _ := newJobFromMap(queue, jobMap)
		queue.Drained = false
		jobChan <- job
		getNextJob(queue, jobChan)
	} else {
		queue.Drained = true
		getNextJob(queue, jobChan)
	}
}
